package com.zhao.algorithm.code

import java.sql.Connection
import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import com.zhao.algorithm.common.{Common, Constant}
import com.zhao.algorithm.utils.{JDBCUtil, OffsetUtils, OpRedis, TimeTransformUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.json.JSONObject
import org.slf4j.{Logger, LoggerFactory}
import redis.clients.jedis.Jedis

/**
 * Description: 用Redis做中间存储,达到UpdateStateByKey效果,同时写入Mysql,并向Kafka发送一条即时消息做触发<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/11/23 14:45
 *需求:
 * 1.本次是为了重构ctr算法,不再按照10分钟一次计算结果,而是从头开始计算,达到某一阈值,即往下写数据
 * 2.具体规则:报告/浏览/分享/站外
 *    1.曝光UV达到10,则触发一次写入Mysql exptid
 *    2.浏览总UV达到10,则触发一次写入Mysql browse
 *    3.分享PV达到1,则触发一次写入Mysql share_inapp
 *    4.分享UV达到1,则触发一次写入Mysql share_inapp
 *    5.站外UV达到10,则触发一次写入Mysql obrowse
 * 3. 2020-11-09 添加:插入mysql时,需要拿出该tid,gid对应的上次的数值,计算出差值
 *
 * 实现:
 * 1.通过redis的string/incr和set集合来分别实现各种行为的uv,pv
 *    1.1 redis的key存储为:tid:gid:exptid_pv / tid:gid:exptid_uv 分别代表tid gid 曝光pv 曝光uv
 *    1.2 取值分别为get:获取pv值 / scard:获取set长度 取到对应结果
 *
 * 2.代码中,针对不同action设置不同阈值,插入前后分别从redis取值
 * 1)去插入前是判断和插入后的相同的不插入(set集合特性)
 * 2)取插入后是判断到阈值需要写入mysql(触发器需求)
 *
 * 3.针对新需求,把mysql读写的操作封装函数,而读出历史数据之后,计算逻辑也封装成函数调用传参即可.
 *
 * @author 柒柒
 * @version : 1.0
 */

object SparkReadKafka2RedisApp {
  @transient
  val logger: Logger = LoggerFactory.getLogger(this.getClass)

  def main(args: Array[String]): Unit = {

    //    if (args.length != 2) {
    //      sys.error(
    //        """
    //          |请输如正确的参数：
    //          |  它的参数需要如下：
    //          |  <seconds> 程序批次间隔时间
    //          |  <select>  redis 数据库编号
    //          |""".stripMargin)
    //      System.exit(1)
    //    }

    val conf = new SparkConf()
      .setAppName("SparkReadKafka2RedisApp")
    //      .setMaster("local[*]")

    val spark: SparkContext = new SparkContext(conf)

    //    spark.setLogLevel("ERROR")

    val ssc = new StreamingContext(spark, Seconds(2))

    val load: Config = ConfigFactory.load("application.properties")

    val appId = "SparkReadKafka2RedisApp"
    val groupId = load.getString("algorithm.kafka.consumer.groupId")
    // 读取历史偏移量（在Driver端查询历史偏移量）
    val offsets: Map[TopicPartition, Long] = OffsetUtils.queryHistoryOffsetFromMySQL(appId, groupId)

    // 获取到使用offset读取的DStream
    val kafkaTopicDS: InputDStream[ConsumerRecord[String, String]] = CustomKafkaConsumer.getKafkaTopicDS(ssc, load, offsets)


    val kafkaProducer: Broadcast[CustomKafkaSink[String, String]] = {
      val kafkaProducerConfig = {
        val props = new Properties()
        props.setProperty("bootstrap.servers", load.getString("algorithm.kafka.producer.bootstrap.servers"))
        //客户端名称 设置之后不能并行化，会报错，多线程下 client_id 默认与 线程号 有关 ，但是设置之后就固定了，不能多线程
        //            props.setProperty("client.id", load.getString("algorithm.kafka.producer.client.id"))
        props.setProperty("key.serializer", classOf[StringSerializer].getName)
        props.setProperty("value.serializer", classOf[StringSerializer].getName)
        props
      }
      logger.info("kafka producer init done!")
      ssc.sparkContext.broadcast(CustomKafkaSink[String, String](kafkaProducerConfig))
    }


    kafkaTopicDS.foreachRDD(rdd => {
      if (!rdd.isEmpty()) {
        // 获取该rdd的offset
        val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.foreachPartition(rddPartition => {
          // 从redis连接池获取连接
          var jedis: Jedis = null
          // 从jdbc连接池获取连接
          var connection: Connection = null
          try {
            // 从redis连接池获取连接
            jedis = OpRedis.getJedis()
            jedis.auth(OpRedis.load.getString("redis.password"))
            jedis.select(10)

            // 从jdbc连接池获取连接
            connection = JDBCUtil.getConn()

            rddPartition.foreach(x => {

              if (x.offset() % 100000 == 0) {
                //                println("当前系统时间是: " + TimeTransformUtil.getNowTime() + ", 从kafka中获取的分区: " + x.partition() + ", 从kafka中获取的offset: " + x.offset())
                logger.info("当前系统时间是: " + TimeTransformUtil.getNowTime() + ", 从kafka中获取的分区: " + x.partition() + ", 从kafka中获取的offset: " + x.offset())
              }

              // 构造无用json，防止log日志传过来的json无效，导致程序异常终止。
              var obj: JSONObject = new JSONObject("{\"message\":\"使用默认的空json字符串\"}")
              try {
                // 转化为json对象
                obj = new JSONObject(x.value())
              } catch {
                case e: Exception => {
                  // throw new RuntimeException 这种有问题，抛出这个异常之后，如果jvm处理不了，还是会导致服务停止
                  //                  throw new RuntimeException("json对象获取失败", e)
                  e.printStackTrace()
                }
              }


              var key1 = ""
              var key2 = ""
              var tid_gid = ""
              var flag = 0
              var log_time = 0
              var action = ""
              var device = ""
              var tid = 0
              var gid = 0

              if (obj.has("action") && obj.has("gid") && obj.has("tid") && obj.has("device") && obj.has("post_time") && obj.has("timestamp") && obj.getString("post_time").toInt >= 1604160000) {

                try {
                  // 获取日志时间
                  log_time = Common.getTime(obj.getString("timestamp")).toInt

                  action = obj.getString("action")
                  device = obj.getString("device")
                  tid = obj.getInt("tid")
                  gid = obj.getInt("gid")
                  tid_gid = tid + ":" + gid
                } catch {
                  case e: Exception => {
                    // throw new RuntimeException 这种有问题，抛出这个异常之后，如果jvm处理不了，还是会导致服务停止
                    //                    throw new RuntimeException("JSON获取时候类型转换有异常", e)
                    e.printStackTrace()
                  }
                }


                //1. 曝光UV/PV
                if (action.equals("exptid")) {

                  // PV
                  key1 = tid + ":" + gid + Constant.EXPTID_PV
                  // UV
                  key2 = tid + ":" + gid + Constant.EXPTID_UV
                  // flag 记录写入redis之前的值，做标记值，以便对比
                  flag = jedis.scard(key2).toInt

                  // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                  Common.addPV2Redis(jedis, key1)
                  Common.addUV2Redis(jedis, key2, device)

                  // 再去判断其增长是否超过阀值，超过阀值则写入MySQL
                  val isThreshold: Boolean = Common.isThreshold(jedis, key2, 10)

                  // 如果达到阀值，则需要获取该KEY的其他Value长度或者结果，做PV/UV，写入MySQL
                  if (isThreshold && flag != jedis.scard(key2)) {

                    // 1.1 先获取该gid tid 对应的其他action的value
                    val restArray: Array[Int] = Common.getOtherPVUV(jedis, tid_gid)

                    val sql_last: String =
                      s"""
                         |select * from pre_exptid_update
                         |where tid=$tid and gid=$gid
                         |order by flush_id desc
                         |limit 1
                         |""".stripMargin

                    // 1.2 再获取该gid tid 对应历史的value
                    val last_arr: Array[Int] = Common.getLastPVUV(connection, sql_last, tid, gid)

                    val sql_insert: String =
                      """
                        |insert into pre_exptid_update(
                        |tid, gid, dtime,
                        |exp_pv, exp_uv, web_pv, web_uv, browse_pv, browse_uv,
                        |browse_all_pv, browse_all_uv, share_pv, share_uv, waken_pv, waken_uv,
                        |exp_pv_inc, exp_uv_inc, web_pv_inc, web_uv_inc, browse_pv_inc, browse_uv_inc,
                        |browse_all_pv_inc, browse_all_uv_inc,share_pv_inc, share_uv_inc,waken_pv_inc,waken_uv_inc,
                        |flush_id)
                        |values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                        |""".stripMargin
                    // 1.3 将 step1 step2 结果进行计算结合，写入mysql
                    Common.execInsertSql(connection, sql_insert, tid, gid, log_time, restArray, last_arr)

                    val json: String = Common.buildJSON(1, tid, gid, log_time, restArray, last_arr)
                    kafkaProducer.value.send("topic_recommend_data_exposureuv", json)

                  }

                }
                //2. 总浏览UV/PV
                else if (action.equals("browse")) {

                  // 2.1 过滤掉站外的浏览。
                  if (obj.has("refer") && !obj.getString("refer").equals("notice") && obj.has("from") && !obj.getString("from").equals("magic-window")) {

                    // PV
                    key1 = tid + ":" + gid + Constant.BROWSE_PV
                    // UV
                    key2 = tid + ":" + gid + Constant.BROWSE_UV

                    // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                    Common.addPV2Redis(jedis, key1)
                    Common.addUV2Redis(jedis, key2, device)

                  }

                  // PV
                  key1 = tid + ":" + gid + Constant.BROWSE_ALL_PV
                  // UV
                  key2 = tid + ":" + gid + Constant.BROWSE_ALL_UV

                  flag = jedis.scard(key2).toInt

                  // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                  Common.addPV2Redis(jedis, key1)
                  Common.addUV2Redis(jedis, key2, device)

                  // 再去判断其增长是否超过阀值，超过阀值则写入MySQL
                  val isThreshold: Boolean = Common.isThreshold(jedis, key2, 10)

                  // 如果达到阀值，则需要获取该KEY的其他Value长度或者结果，做PV/UV，写入MySQL
                  if (isThreshold && flag != jedis.scard(key2)) {

                    // 2.1 先获取该gid tid 对应的其他action的value
                    val restArray: Array[Int] = Common.getOtherPVUV(jedis, tid_gid)

                    val sql_last: String =
                      s"""
                         |select * from pre_browse_all_update
                         |where tid=$tid and gid=$gid
                         |order by flush_id desc
                         |limit 1
                         |""".stripMargin

                    // 2.2 再获取该gid tid 对应历史的value
                    val last_arr: Array[Int] = Common.getLastPVUV(connection, sql_last, tid, gid)

                    val sql_insert =
                      """
                        |insert into pre_browse_all_update(
                        |tid, gid, dtime,
                        |exp_pv, exp_uv, web_pv, web_uv, browse_pv, browse_uv,
                        |browse_all_pv, browse_all_uv, share_pv, share_uv, waken_pv, waken_uv,
                        |exp_pv_inc, exp_uv_inc, web_pv_inc, web_uv_inc, browse_pv_inc, browse_uv_inc,
                        |browse_all_pv_inc, browse_all_uv_inc,share_pv_inc, share_uv_inc,waken_pv_inc,waken_uv_inc,
                        |flush_id )
                        |values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                        |""".stripMargin

                    // 2.3 将 step1 step2 结果进行计算结合，写入mysql
                    Common.execInsertSql(connection, sql_insert, tid, gid, log_time, restArray, last_arr)


                    val json: String = Common.buildJSON(2, tid, gid, log_time, restArray, last_arr)
                    kafkaProducer.value.send("topic_recommend_data_viewuv", json)

                  }

                }
                //3. 分享UV + 分享PV
                else if (action.equals("share_inapp")) {

                  // PV
                  key1 = tid + ":" + gid + Constant.SHARE_PV
                  // UV
                  key2 = tid + ":" + gid + Constant.SHARE_UV

                  flag = jedis.scard(key2).toInt

                  // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                  Common.addPV2Redis(jedis, key1)
                  Common.addUV2Redis(jedis, key2, device)


                  // 3.1 share pv 写入一个库


                  // 3.1.1 先获取该gid tid 对应的其他action的value
                  val restArray: Array[Int] = Common.getOtherPVUV(jedis, tid_gid)

                  val sql_last: String =
                    s"""
                       |select * from pre_share_pv_update
                       |where tid=$tid and gid=$gid
                       |order by flush_id desc
                       |limit 1
                       |""".stripMargin

                  // 3.1.2 再获取该gid tid 对应历史的value
                  val last_arr: Array[Int] = Common.getLastPVUV(connection, sql_last, tid, gid)


                  val sql_insert =
                    """
                      |insert into pre_share_pv_update(
                      |tid, gid, dtime,
                      |exp_pv, exp_uv, web_pv, web_uv, browse_pv, browse_uv,
                      |browse_all_pv, browse_all_uv, share_pv, share_uv, waken_pv, waken_uv,
                      |exp_pv_inc, exp_uv_inc, web_pv_inc, web_uv_inc, browse_pv_inc, browse_uv_inc,
                      |browse_all_pv_inc, browse_all_uv_inc,share_pv_inc, share_uv_inc,waken_pv_inc,waken_uv_inc,
                      |flush_id )
                      |values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                      |""".stripMargin
                  // 3.1.3 将 step1 step2 结果进行计算结合，写入mysql
                  Common.execInsertSql(connection, sql_insert, tid, gid, log_time, restArray, last_arr)

                  val json: String = Common.buildJSON(3, tid, gid, log_time, restArray, last_arr)
                  kafkaProducer.value.send("topic_recommend_data_sharepv", json)

                  // 3.2 share uv 写入一个库

                  // 再get一遍uv和写入reids之前的value判断，相差为1 则uv增加
                  if (flag == 0 || jedis.scard(key2) - flag == 1) {

                    // 这里是写入share uv 增1的那个表
                    // 3.2.1 先获取该gid tid 对应的其他action的value
                    val restArray: Array[Int] = Common.getOtherPVUV(jedis, tid_gid)

                    val sql_last: String =
                      s"""
                         |select * from pre_share_uv_update
                         |where tid=$tid and gid=$gid
                         |order by flush_id desc
                         |limit 1
                         |""".stripMargin

                    // 3.2.2 再获取该gid tid 对应历史的value
                    val last_arr: Array[Int] = Common.getLastPVUV(connection, sql_last, tid, gid)

                    val sql_insert =
                      """
                        |insert into pre_share_uv_update(
                        |tid, gid, dtime,
                        |exp_pv, exp_uv, web_pv, web_uv, browse_pv, browse_uv,
                        |browse_all_pv, browse_all_uv, share_pv, share_uv, waken_pv, waken_uv,
                        |exp_pv_inc, exp_uv_inc, web_pv_inc, web_uv_inc, browse_pv_inc, browse_uv_inc,
                        |browse_all_pv_inc, browse_all_uv_inc,share_pv_inc, share_uv_inc,waken_pv_inc,waken_uv_inc,
                        |flush_id )
                        |values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                        |""".stripMargin

                    // 3.2.3 将 step1 step2 结果进行计算结合，写入mysql
                    Common.execInsertSql(connection, sql_insert, tid, gid, log_time, restArray, last_arr)

                    val json: String = Common.buildJSON(4, tid, gid, log_time, restArray, last_arr)
                    kafkaProducer.value.send("topic_recommend_data_shareuv", json)

                  }


                }
                //4. 唤醒 (不需要触发器)
                else if (action.equals("waken")) {
                  // 写入VALUE计算PV
                  key1 = tid + ":" + gid + Constant.WAKEN_PV
                  // 写入SET集合计算UV
                  key2 = tid + ":" + gid + Constant.WAKEN_UV

                  // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                  Common.addPV2Redis(jedis, key1)
                  Common.addUV2Redis(jedis, key2, device)

                }
                //5. 站外UV
                else if (action.equals("obrowse")) {

                  // 如果有标记短信字段，则去除，如果没有，则添加
                  if ((obj.has("browser") && !obj.get("browser").equals("browser")) || (!obj.has("browser"))) {

                    // PV
                    key1 = tid + ":" + gid + Constant.OBROWSE_PV
                    // UV
                    key2 = tid + ":" + gid + Constant.OBROWSE_UV

                    flag = jedis.scard(key2).toInt

                    // 不管是否需要写入MySQL，至少要先保存到Redis中。再其在MySQL的判断长度。
                    Common.addPV2Redis(jedis, key1)
                    Common.addUV2Redis(jedis, key2, device)

                    // 再去判断其增长是否超过阀值，超过阀值则写入MySQL
                    val isThreshold: Boolean = Common.isThreshold(jedis, key2, 10)

                    // 如果达到阀值，则需要获取该KEY的其他Value长度或者结果，做PV/UV，写入MySQL
                    if (isThreshold && flag != jedis.scard(key2)) {
                      // 5.1 再获取该gid tid 对应历史的value
                      val restArray: Array[Int] = Common.getOtherPVUV(jedis, tid_gid)

                      val sql_last: String =
                        s"""
                           |select * from pre_web_update
                           |where tid=$tid and gid=$gid
                           |order by flush_id desc
                           |limit 1
                           |""".stripMargin

                      // 5.2 再获取该gid tid 对应历史的value
                      val last_arr: Array[Int] = Common.getLastPVUV(connection, sql_last, tid, gid)


                      val sql_insert =
                        """
                          |insert into pre_web_update(
                          |tid, gid, dtime,
                          |exp_pv, exp_uv, web_pv, web_uv, browse_pv, browse_uv,
                          |browse_all_pv, browse_all_uv, share_pv, share_uv, waken_pv, waken_uv,
                          |exp_pv_inc, exp_uv_inc, web_pv_inc, web_uv_inc, browse_pv_inc, browse_uv_inc,
                          |browse_all_pv_inc, browse_all_uv_inc,share_pv_inc, share_uv_inc,waken_pv_inc,waken_uv_inc,
                          |flush_id )
                          |values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                          |""".stripMargin

                      // 5.3 将 step1 step2 结果进行计算结合，写入mysql
                      Common.execInsertSql(connection, sql_insert, tid, gid, log_time, restArray, last_arr)

                      //                    val json: String = Common.buildJSON(5, tid, gid, log_time, restArray, last_arr)
                      //                    kafkaProducer.value.send("topic_recommend_data_webuv", json)

                    }

                  }

                }
                //6. 不是上述五种Action，则归为此
                //            else {
                //              print("不需要的action，过滤掉！")
                //            }
              }
              //          // 1.1 action/gid/tid/ 任意一个获取不到，则归为此
              //          else {
              //            keys = "0-0-0"
              //          }

            })
          } catch {
            case e: Exception => {
              // throw new RuntimeException 这种有问题，抛出这个异常之后，如果jvm处理不了，还是会导致服务停止
              //              throw new RuntimeException("主程序运行异常：", e)
              e.printStackTrace()
            }
          }
          finally {
            // 关闭jdbc资源和redis资源
            if (connection != null) {
              JDBCUtil.releaseCon(connection)
            }
            if (jedis != null) {
              jedis.close()
            }
          }
        })

        // 手动提交offset 至 mysql
        OffsetUtils.sendOffsetToMySQL(ssc, appId, groupId, ranges)
      }

    })


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}






















